package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}

object StudentScoresAdvanced {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Student Scores Advanced").setMaster("local[4]") // 设置本地运行环境，使用4个线程
    val sc = new SparkContext(conf)

    // 假设我们有如下格式的学生数据：(name, class, score)
    val studentData = Seq(
      ("Alice", "Class1", 89),
      ("Bob", "Class2", 76),
      ("Charlie", "Class1", 93),
      ("David", "Class2", 85),
      ("Eva", "Class1", 90),
      ("Frank", "Class3", 78),
      ("Grace", "Class3", 92)
    )

    // 将本地集合并行化为RDD，并设置足够的分区数以确保多个tasks
    val studentsRDD = sc.parallelize(studentData, 4)

    // 将RDD保存到本地文件系统
    studentsRDD.saveAsTextFile("/opt/code/output/students")

    // 计算每个班级学生总成绩
    val totalScoresPerClass = studentsRDD
      .map { case (_, className, score) => (className, score) } // 转换为(班级, 成绩)对
      .reduceByKey(_ + _) // 对每个班级的成绩求和

    // 找出所有班级中最高总成绩
//    val maxTotalScore = totalScoresPerClass
//      .map { case (_, totalScore) => totalScore } // 提取总成绩
//      .reduce(math.max) // 找到最高的总成绩
//
//    println(s"Max Total Score across all classes: $maxTotalScore")

    // 触发action操作，例如collect()，以确保前面的操作被执行
    totalScoresPerClass.collect().foreach { case (className, totalScore) =>
      println(s"$className: $totalScore")
    }

    // 在停止SparkContext之前暂停5分钟（300000毫秒），给您时间查看Spark UI
    Thread.sleep(300000)

    sc.stop()
  }
}